package com.codeduck.kafka.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;

/**
 * 根据topic.properties创建Topic
 *
 * @auther Mahone Yan
 * @date 2023/5/7
 */
@Component
public class TopicConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(TopicConfig.class);
    private static final String RESOURCE_PATH = "kafka/topic.properties";
    private static final String KAFKA_SERVER = "bootstrap.servers";
    private final Properties props = new Properties();

    public void init() {
        initTopicProperties();
        createTopics();
        LOGGER.info("=====================> Kafka Topic 初始化完毕 <=====================");
    }

    /**
     * 创建Topic
     */
    private void createTopics() {
        Map<String, Object> config = new HashMap<>();
        config.put(KAFKA_SERVER, props.get(KAFKA_SERVER));
        AdminClient adminClient = AdminClient.create(config);
        adminClient.createTopics(getTopics());
    }

    /**
     * 解析获取Topic列表
     */
    private List<NewTopic> getTopics() {
        List<NewTopic> topics = new ArrayList<>(props.size());
        props.forEach((key, val) -> {
            if (StringUtils.isNotBlank(val.toString()) && val.toString().trim().contains(":")) {
                String[] info = val.toString().trim().split(":");
                if (info.length < 3) {
                    return;
                }
                try {
                    NewTopic newTopic = new NewTopic(info[0], Integer.parseInt(info[1]), Short.parseShort(info[2]));
                    topics.add(newTopic);
                } catch (NumberFormatException e) {
                    LOGGER.error("[{}]配置不符合约定，将跳过Topic的创建", val, e);
                }
            }
        });
        return topics;
    }

    /**
     * 加载Topic配置文件
     */
    public void initTopicProperties() {
        InputStream inputStream = TopicConfig.class.getClassLoader().getResourceAsStream(RESOURCE_PATH);
        try {
            props.load(inputStream);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
